package com.vteba.shards.schema.service;

import org.I0Itec.zkclient.ZkClient;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.InitializingBean;

import com.vteba.utils.common.Assert;
import com.vteba.utils.common.PropUtils;

public class SchemaInfoListener implements InitializingBean {
	private static final Logger LOGGER = LogManager.getLogger(SchemaInfoListener.class);
	
	private SchemaInfoService schemaInfoService;
	private ZkClient zkClient;
	
	public SchemaInfoListener() {
		super();
	}

	public SchemaInfoListener(SchemaInfoService schemaInfoService) {
		super();
		this.schemaInfoService = schemaInfoService;
	}

	/**
	 * 初始化ZKClient
	 */
	public void init() {
		// 没有注入，才去初始化
		if (zkClient == null) {
			String serverList = PropUtils.get("zookeeper.server");
			Assert.notNull(serverList, "config.properties lose 'zookeeper.server'.");
			
			int connectionTimeout = PropUtils.getInt("zookeeper.timeout", 100000);
			zkClient = new ZkClient(serverList, connectionTimeout);
			if (LOGGER.isInfoEnabled()) {
				LOGGER.info("zookeeper server [{}] connected success.", serverList);
			}
		}
		
		SchemaTask task = new SchemaTask(zkClient, schemaInfoService);
		Thread thread = new Thread(task);
		thread.start();
	}

	public void setSchemaInfoService(SchemaInfoService schemaInfoService) {
		this.schemaInfoService = schemaInfoService;
	}

	public ZkClient getZkClient() {
		return zkClient;
	}

	public void setZkClient(ZkClient zkClient) {
		this.zkClient = zkClient;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		Assert.notNull(schemaInfoService, "property [schemaInfoService] cannot be null.");
		init();
	}
}
